package cn.ant.Client;

import cn.ant.utils.ChannelUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * 消费者（正常的消息消费）；若消费消息失败则转发到死信交换机中
 *
 * @author Anhui AntLaddie <a href="https://juejin.cn/user/4092805620185316">(掘金蚂蚁小哥)</a>
 * @version 1.0
 **/
public class Consumer {

    // 通过日志管理器获取Logger对象
    static Logger logger = LogManager.getLogger(Consumer.class);

    // 直接交换机名称
    public static final String NORMAL_EXCHANGE = "NormalExchange";
    // 队列名称
    public static final String NORMAL_QUEUE = "NormalQueue";
    // 路由Key（Routing Key）
    public static final String NORMAL_KEY = "NormalKey";

    // 声明死信交换机名称
    public static final String DLX_EXCHANGE = "DLXExchange";
    // 声明死信交换机到死信队列路由绑定关系（Routing Key）
    public static final String DLX_KEY = "DLXKey";

    /***
     * 正常的消费者，用来消息消息
     */
    public static void main(String[] args) throws IOException {
        // 通过工具类获取一个信道
        Channel channel = ChannelUtil.getChannel();
        // 创建名称为"NormalExchange"的直接交换机；防止消费者先启动报错，找不到交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT,
                true, false, false, null);

        // ~~~~~~~~~~~~~~~~~~~~~~~~~~ 设置队列配置参数Start ~~~~~~~~~~~~~~~~~~~~~~~~~~
        Map<String, Object> arguments = new HashMap<>();
        // 正常队列设置死信交换机 参数key是固定值；（就是说死去的消息发送到哪个交换机）
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE);
        // 正常队列设置死信交换机到死信队列绑定的 Routing Key，参数key是固定值；
        // 就是说死去的消息在交换机里通过什么路由发送到死信队列
        arguments.put("x-dead-letter-routing-key", DLX_KEY);
        // 设置正常队列的长度限制 为3（## 队列达到最大长度产生死信）
        // arguments.put("x-max-length",3);
        // ~~~~~~~~~~~~~~~~~~~~~~~~~~  设置队列配置参数End  ~~~~~~~~~~~~~~~~~~~~~~~~~~

        // 创建一个队列（其中arguments配置参数里面设置了死信交换机）
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        // 把队列绑定到指定交换机上
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, NORMAL_KEY);
        logger.info("消费者（TTLConsumer）开始监听队列消息....");

        // ## 消息TTL过期产生死信
        // 若不写下面的从队列中取出消息消费的代码，那么就等消息在队列中自动过期，消息10秒过期（生产者中设置）

         // ## 消息被拒产生死信
         // 接收消息
         channel.basicConsume(NORMAL_QUEUE, false, (consumerTag, message) -> {
             //## 消息处理（处理方式：若处理的消息内包含 "b" 字符串则拒绝处理，并设置手动不确认，并丢弃消息 ）
             //获取消息体
             String msg = new String(message.getBody(), StandardCharsets.UTF_8);
             logger.info("消费者获取队列信息正在处理：{}", msg);
             if (msg.contains("b")) {
                 //拒绝处理
                 logger.info("=======》 X 消息已被拒绝处理，并丢弃：{}", msg);
                 channel.basicNack(message.getEnvelope().getDeliveryTag(), false, false);
             } else {
                 logger.info("=======》 √ 消息已被成功处理：{}", msg);
                 channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
             }
         }, consumerTag -> {
             logger.info("监听的队列出现异常；可能队列被删除！{}", consumerTag);
         });
    }
}
